/**
 * 
 */
package pt.inescid.components.overlaymanager;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.Vector;
import java.util.Collections;

import pt.inescid.components.api.Component;
import pt.inescid.components.api.OverlayManagerI;
import pt.inescid.components.commons.Resources;
import pt.inescid.components.messages.ACKMsg;
import pt.inescid.components.messages.AvailabilityMsg;
import pt.inescid.components.messages.Task;
import pt.inescid.components.messages.TaskResult;
import pt.inescid.components.messages.UpdateMsg;
import pt.inescid.core.GridP2P;
import pt.inescid.utils.FileUtils;
import pt.inescid.utils.Linpack;
import pt.inescid.utils.Log;


import rice.Continuation;
import rice.environment.Environment;
import rice.p2p.commonapi.Id;
import rice.p2p.commonapi.Message;
import rice.p2p.commonapi.NodeHandle;
import rice.p2p.past.Past;
import rice.p2p.past.PastContent;
import rice.p2p.past.PastImpl;
import rice.pastry.NodeIdFactory;
import rice.pastry.PastryNode;
import rice.pastry.PastryNodeFactory;
import rice.pastry.commonapi.PastryIdFactory;
import rice.pastry.socket.SocketPastryNodeFactory;
import rice.pastry.standard.RandomNodeIdFactory;
import rice.persistence.LRUCache;
import rice.persistence.MemoryStorage;
import rice.persistence.Storage;
import rice.persistence.StorageManagerImpl;
import rice.selector.TimerTask;

/** 
 * 
 * @author Filipe Paredes
 * @author Sergio Esteves (sesteves at gsd.inesc-id.pt)
 */
public class OverlayManager extends Component implements OverlayManagerI, Runnable {

	// maximum MFlops/s for the linpack challenge
	protected final static double MAX_CPU = 500.0;
	protected final static double MAX_MEM = 4000.0;

	protected final static int NUM_RETRIES = 200;
	
	protected final static long TIMEOUT = 5000;
	
	public int NUM_NEIGHBORS = 62;
	
	private String tempDir;
	
	private long maxStorage;
	
	private long updateInterval;
	private Linpack linpack = new Linpack();
	
	private Environment env = new Environment();

	private int waitingNReplies;
	private int nAvailableNodes;
	private int notAvailableNodes;
	private Object availableLock = new Object();
	
	// Our pastry node
	private PastryNode node;
	
	// Machine Own Resource Description
	private Resources resourceManif = new Resources();
		
	private boolean networkInitialized = false;

	private long seqNumber;
	
	// Buffer of update messages arrived before node initialization is complete
	private Hashtable<Id,List<UpdateMsg>> updatesBuffer = 
		new Hashtable<Id,List<UpdateMsg>>();

	private Hashtable<Id,NodeMetrics> neighborsRate =
		new Hashtable<Id,NodeMetrics>();;
 
	// contains the reputation of surrounding nodes
	private Hashtable<Id,NodeReputation> nodesReputation = 
		new Hashtable<Id,NodeReputation>();
	
	//FIXME
	// available nodes chosen to forward a job, NodeMetrics for reducing its availability
	//private Hashtable<Id,NodeMetrics> nodesReceivers;

	// List of chosen nodes to forward a certain message (avoids repeating)
	private List<Id> nodesForwarded;
	
	
	private List<Id> randomChoice;

	// Stores available nodes that messages are forwarded	
	private Hashtable<Integer,List<Id>> chosenAvailableNodes =
		new Hashtable<Integer,List<Id>>();

	private Past pastApp;
	
	private byte[] result;
	private boolean resultDone;
	
	private Map<Id,Long> updateTime = new Hashtable<Id,Long>();
	
	/**
	 * 
	 * @param GridP2P
	 * @param updateInterval
	 * @param tempDir
	 * @param maxStorage
	 */
	public OverlayManager(GridP2P gridP2P, long updateInterval, String tempDir, long maxStorage) {						
		super(gridP2P);
		this.updateInterval = updateInterval;
		this.tempDir = tempDir;
		this.maxStorage = maxStorage;
		log.write(this, "OverlayManager constructed", Log.FINER);
	}
	
	private void initializePast() {
		log.write(this, "initializePast called", Log.FINER);
		
		PastryIdFactory idf = new rice.pastry.commonapi.PastryIdFactory(env);
		Storage stor = new MemoryStorage(idf);
		pastApp = new PastImpl(node, new StorageManagerImpl(idf, stor, 
				new LRUCache(new MemoryStorage(idf), 512 * 1024, node.getEnvironment())),
				0, "");	      
	}
	
	/**
	 * Initializes the node
	 * 
	 * @param bindPort the local port to bind to
	 * @param bootIP the IP of the node to boot from
	 * @param bootPort the port of the node to boot from
	 */
	public void init(int bindPort, String bootIP, int bootPort) {
		log.write(this, "init called (bindPort: " + bindPort + ", bootIP: " + bootIP +
				", bootPort: " + bootPort + ")", Log.FINER);
				
		updateResourceManifest();		
		try {
			InetAddress bootaddr = InetAddress.getByName(bootIP);
			InetSocketAddress bootaddress = new InetSocketAddress(bootaddr,bootPort);
	
			// Generate the NodeIds Randomly
			NodeIdFactory nidFactory = new RandomNodeIdFactory(env);

			// construct the PastryNodeFactory
			PastryNodeFactory factory = new SocketPastryNodeFactory(nidFactory, bindPort, env);
			
			node = factory.newNode();			
			gridP2P.registerEndPoint();
			
			//FIXME
			this.networkInitialized = true;
							
			node.boot(bootaddress);
			
			// the node may require sending several messages to fully boot into the ring
			synchronized(node) {
				while(!node.isReady() && !node.joinFailed()) {
					// delay so we don't busy-wait
					node.wait(500);

					// abort if can't join
					if (node.joinFailed()) {
						log.write(this, "Could not join the FreePastry ring: " +
								node.joinFailedReason(), Log.SEVERE);
						//FIXME
						//throw new IOException("Could not join the FreePastry ring.  Reason:"+node.joinFailedReason()); 
					}
				}       
			}	
			log.write(this, "Finished creating new node " + node, Log.SEVERE);																			
			initializePast();
			// start thread for sending update messages periodically 
			new Thread(this).start();
			
		} catch(UnknownHostException e) {
			log.write(this, "UnknowHostException occurred: " + e.getMessage(), Log.SEVERE);	
		} catch(IOException e) {
			log.write(this, "IOException occurred: " + e.getMessage(), Log.SEVERE);			
		} catch(InterruptedException e) {
			log.write(this, "InterruptedException occurred: " + e.getMessage(), Log.SEVERE);			
		}		
	}
	
	public boolean networkInitialized() {
		return networkInitialized;
	}
	
	private long getAvailableStorage() {
		long used = FileUtils.getDirSize(tempDir);		
		return maxStorage - used;
	}
	
	private void updateResourceManifest() {
		log.write(this, "updateResourceManifest called", Log.FINER);		
		
		double cpu = 0, cputmp;
		for(int i = 0; i < 3; i++) {
			cputmp = linpack.run_benchmark();		
			if(cputmp > cpu) 
				cpu = cputmp;
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				log.write(this, "InterruptedException occurred: " + e.getMessage(), Log.FINER);
			}
		}
						
		double mem = Runtime.getRuntime().freeMemory() / 1024;
		double bw = 0.5;
		long storage = getAvailableStorage();
	
		// normalizing values
		cpu = cpu > MAX_CPU ? 1.0 : cpu / MAX_CPU;
		mem = mem > MAX_MEM ? 1.0 : mem / MAX_MEM;

		log.write(this, "Resources (cpu: " + cpu + ", free mem: " + 
				mem + ", bw: " + bw + ", storage: " + storage + ")", Log.FINER);
		
		synchronized(resourceManif) {
			resourceManif.setAll(cpu, mem, bw, storage);
		}
	}		

	public void updateResources(double cpu, double mem, double bw) {
		log.write(this, "updateResources called (cpu: " + cpu + ", mem: " + mem +
				", bw: " + bw + ")", Log.FINER);
		synchronized(resourceManif) {
			resourceManif.updateCPU(cpu);
			resourceManif.updateMem(mem);
			resourceManif.updateBW(bw);
			resourceManif.setStorage(getAvailableStorage());
		}
		this.sendUpdateMessages();
	}
	
	public void updateResources() {
		updateResourceManifest();
		sendUpdateMessages();
	}
	
	/*
	 * check if a Node is suitable to process a certain task
	 */
	private boolean isSuitable(NodeMetrics nm, Resources taskCost) {

		if(taskCost.getCPU() > nm.getCPU()) return false;
		if(taskCost.getMem() > nm.getMem()) return false;
		if(taskCost.getBW() > nm.getBW()) return false;
		if(taskCost.getStorage() > nm.getStorage()) return false;

		return true;
	}
	
	
	private Map<Double,Id> getSuitableNodes(String appID, Resources taskCost) {
		// the best nodes come first
		Map<Double,Id> suitableNodes = new TreeMap<Double,Id>(Collections.reverseOrder());
		for(NodeMetrics nm : neighborsRate.values()) {
			Id id = nm.getId();
			if(!nm.getNodeHandle().checkLiveness()){				
				nodesReputation.remove(id);				
				randomChoice.remove(id);						
				continue;
			}
			if(!nm.isAvailable()) continue;
			double rate = nm.getRate();
			if(rate < 0.2) continue;
			if(!nm.containsApp(appID)) continue;
			if(!isSuitable(nm, taskCost)) continue;
			while(suitableNodes.containsKey(rate))
				rate += 0.001;
			suitableNodes.put(rate, id);				
		}
		return suitableNodes;
	}

	/**
	 * get the number of available nodes
	 * 
	 * @return number of available nodes
	 */
	public int getNumAvailableNodes(String appID, int limit) {
		if(neighborsRate.isEmpty()) 
			return 0;
		int alpha;
		synchronized(availableLock) {
			alpha = notAvailableNodes;
			notAvailableNodes = 0;
			waitingNReplies = 0;
			nAvailableNodes = 0;
		}
		Map<Double,Id> bestNodes = new TreeMap<Double,Id>(Collections.reverseOrder());
		for(NodeMetrics nm : neighborsRate.values()) {
			Id id = nm.getId();
			if(!nm.getNodeHandle().checkLiveness()){
				nodesReputation.remove(id);				
				randomChoice.remove(id);						
				continue;
			}
			
			if(!gridP2P.precontrolOutbound(nm)) {
				nm.setAvailability(false);
				continue;
			}
			double rate = nm.getRate();
			
			// minimum QoS = 20%
			if(rate < 0.2) continue;
			if(!nm.containsApp(appID)) continue;			
			
			while(bestNodes.containsKey(rate))
				rate += 0.001;						
			bestNodes.put(rate, id);			
		}
		
		AvailabilityMsg m = new AvailabilityMsg(node.getLocalNodeHandle(), 
				AvailabilityMsg.IS_AVAILABLE);
		gridP2P.controlOutbound(m);			
		for(Id id : bestNodes.values()) {
			gridP2P.routeMyMsg(id, m);
			synchronized(availableLock) {
				if(++waitingNReplies == limit + alpha) break;
			}
		}
		
		synchronized(this) {
			try {
				this.wait(TIMEOUT);
			} catch (InterruptedException e) {			
			}		
		}
		return nAvailableNodes > limit ? limit : nAvailableNodes;
	}

	public int distributeTasks(String appID, List<Task> tasks, Resources taskCost) {
		log.write(this, "distributeTasks called (appID: " + appID + 
				", tasks: " + tasks + ", taskCost: " + taskCost + ")", Log.FINER);
		
		int nTasks = tasks.size();
		while(nTasks > 0) {
			Map<Double,Id> suitableNodes = getSuitableNodes(appID, taskCost);
			for(Id id : suitableNodes.values()) {
				gridP2P.routeMyMsg(id, tasks.get(--nTasks));
				if(nTasks == 0) break;
			}
		}
		return 0;
	}

	public int sendResult(Id id, int jobHandler, int taskNumber, byte[] output, double workDone) {		
		TaskResult message = new TaskResult(jobHandler, taskNumber, output, workDone);
		
		// sending to the job holder node
		gridP2P.routeMyMsg(id, message);
		gridP2P.controlOutbound(message);		
		return 0;
	}
	
	//TODO refactor this method!!!
	/**
	 * Selects the most probable available node for processing a job, routing the message to it 
	 * 
	 * @param id - job sender's id
	 * @param message - message (job) to send
	 */
	public synchronized Id forwardMessage(Id id, Message message) {
		log.write(this, "forwardMessage called (id: " + id + ", message: " + 
				message + ")", Log.FINER);
		
		Id idDest = null;
		double lim = 0;

		List<Id> deadNodes = new ArrayList<Id>();
		synchronized (neighborsRate) {

			for(NodeMetrics nm : neighborsRate.values()) {

				Id neighborId = nm.getId();

				//FIXME .isAlive() was used instead
				if(!((NodeMetrics)neighborsRate.get(neighborId)).getNodeHandle().checkLiveness()){
					deadNodes.add(neighborId);
					continue;
				}

				List<Id> lst = chosenAvailableNodes.get(message.hashCode());
				if(lst != null &&  lst.contains(neighborId))
					continue;

				// do not forward to the sender (id)
				if (neighborId.equals(id))
					continue;

				//FIXME
				//long availability = gm.checkAvailability(nm.getParams(), message);
				long availability = 1;

				// Availability and has [best proximity plus more resources]
				if (availability > 0 && nm.getRate() > lim) {
					idDest = neighborId;
					lim = nm.getRate();
				}
			}
		} 

		// Clear dead nodes
		for(Id deadId : deadNodes) {
			synchronized (this.neighborsRate){
				neighborsRate.remove(deadId);
			}
			synchronized (this.nodesReputation){
				nodesReputation.remove(deadId);
			}
			randomChoice.remove(deadId);
		}
		deadNodes.clear();

		// if not null then we must pre-reduced the params of this neighbor node
		if (idDest != null) {
			synchronized (this.chosenAvailableNodes){
				List<Id> lst = chosenAvailableNodes.get(message.hashCode());
				if(lst == null) {
					lst = new ArrayList<Id>();
					chosenAvailableNodes.put(message.hashCode(), lst);
				}
				lst.add(idDest);					
			}

			// Pre-reducing resources and scheduling restore
			
			//FIXME
			//Vector cost = gm.getMessageCost(message);
			Vector<Double> cost = null;

			if(message.getClass() == Task.class)
				//cost = ((Task)message).getCost();
				;//FIXME

			long restoreTime = (((long) this.getProximity(idDest)) * 2) + 5;

			updateNeighborParams(idDest, cost, true);
			scheduleUpdateNeighborParams(idDest, cost, false, restoreTime);
		} else { // choose the best node to forward
			Id idDest_alt = null;
			double lim_alt = 100;
			lim = 100;

			synchronized(nodesReputation){
				for(NodeReputation rep : nodesReputation.values()) {
					if (!nodesForwarded.contains(idDest) && rep.getRate() < lim) {
						idDest = rep.getId();
						lim = rep.getRate();
					}

					if (rep.getAlternative() < lim_alt) {
						idDest_alt = rep.getId();
						lim_alt = rep.getAlternative();
					}
				}

				// Choose based on the best known node of the idDest_alt node
				if (idDest == null)
					idDest = idDest_alt;

			} 
		}

		synchronized(this.nodesReputation) {
			if(idDest != null) {
				nodesForwarded.add(idDest);
				if(nodesForwarded.size() == nodesReputation.keySet().size())
					nodesForwarded.clear();
			}
		}

		// clearly, the local network doesn't have availability, so the most
		// distant neighbor is selected
		if (idDest == null) {
			if(randomChoice.isEmpty()){
				synchronized (this.neighborsRate) {
					for(Id idNeighbor : neighborsRate.keySet())	
						randomChoice.add(idNeighbor);
				}
			}

			idDest = getMostDistantNode(randomChoice);
			if(idDest == null){
				randomChoice.clear();
				idDest = neighborsRate.keySet().iterator().next();
			}
			randomChoice.remove(idDest);
		}

		if (idDest == null) {
			log.write(this, 
					"Couldn't find any more nodes to send to; the forward of this message will stop here", 
					Log.FINE);
			return null;
		}

		gridP2P.routeMyMsg(idDest, message);
		return idDest;
	}

	private void updateNeighborParams(Id id, Vector<Double> cost, boolean consume){
		log.write(this, "updateNeighborParams called (id: " + id + ", cost: " + cost +
				", consume: " + consume + ")", Log.FINER);
		
		synchronized (neighborsRate) {
			if (neighborsRate.containsKey(id)) {
				NodeMetrics nm = neighborsRate.get(id);
				nm.updateMetrics(cost.get(0), cost.get(1), cost.get(2),
						0, null, consume);
			}
		}
	}


	public double getProximity(Id id) {
		log.write(this, "getProximity called (id: " + id + ")", Log.FINER);
		
		if(neighborsRate.containsKey(id))
			return ((NodeMetrics) neighborsRate.get(id)).getProximity();
		else
			return 0;
	}

	private void scheduleUpdateNeighborParams(final Id id, final Vector<Double> cost, 
			final boolean consume, long delay){
		log.write(this, "scheduleUpdateNeighborParams called (id: " + id + ", cost: " +
				cost + ", consume: " + consume + ", delay: " + delay + ")", Log.FINER);

		env.getSelectorManager().getTimer().schedule(new TimerTask() {
			public void run() {
				synchronized (neighborsRate) {
					if (neighborsRate.containsKey(id)) {
						NodeMetrics nm = (NodeMetrics) neighborsRate.get(id);
						nm.updateMetrics(cost.get(0), cost.get(1), cost.get(2),
								0, null, consume);
					}
				}
			}
		}, delay);
	}

	private Id getMostDistantNode(List<Id> nodes) {
		log.write(this, "getMostDistantNode called", Log.FINER);
		
		Id mostDistantNode = null;
		double mostDistant = 0;
			
		for(Id node : nodes) {
			double nodeDistance = getProximity(node);
			if(nodeDistance > mostDistant){
				mostDistantNode = node;
				mostDistant = nodeDistance; 
			}
		}
		return mostDistantNode;
	}

	public void messageReceived(Id id, UpdateMsg message) {
		log.write(this, "updateMsg received (id: " + id +
				", message: " + message + ")", Log.FINER);
		
		if(!networkInitialized) {
			log.write(this, "network not initialized yet", Log.FINER);
			List<UpdateMsg> lst = new ArrayList<UpdateMsg>();
			synchronized (updatesBuffer) {
				if(updatesBuffer.containsKey(id))
					lst = updatesBuffer.get(id);
				lst.add(message);
				updatesBuffer.put(id, lst);
			}
			return;
		}
				
		// Check if is a NACK
		if(message.getSource().equals(node.getId())){
			log.write(this, "NACK message", Log.FINER);
			neighborsRate.remove(id);
			nodesReputation.remove(id);
			return;
		}

		if(!neighborsRate.containsKey(id)) {
			log.write(this, "neighborsRate does not contain the id", Log.FINER);

			if(node.getId().compareTo(id) > 0){
				int prox = gridP2P.getEndpoint().proximity(message.getSourceHandle());
				neighborsRate.put(id, new NodeMetrics(message.getSourceHandle(), 
						prox));
				nodesReputation.put(id, new NodeReputation(id, NUM_RETRIES));

				log.write(this, "neighborsRate: " + neighborsRate, Log.FINER);
				log.write(this, "Sending ACK", Log.FINER);
				// Send ACK
				gridP2P.routeMyMsg(id, buildUpdateMsg());
				gridP2P.routeMyMsg(id, new ACKMsg(node.getLocalNodeHandle()));
			} else {				
				log.write(this, "Sending NACK", Log.FINER);

				// Send NACK
				gridP2P.routeMyMsg(id, message);
				return;
			}
		}
		log.write(this, "neighborsRate contains the id", Log.FINER);						

		NodeMetrics nm = neighborsRate.get(id);
		if(message.getSeqNumber() < nm.getSeqNumber()) {
			log.write(this, "message outdated", Log.FINE);
			return;
		}

		// Clear all restrictions of outdated info about this node
		clearRestrictions(message.getSource());

		if(message.getRestoreTime() > 0) {
			nm.updateMetrics(message.getCpu(), message.getMem(), message.getBW(),
					message.getStorage(), message.getAppIDs(), message.getConsume());
			nm.setSeqNumber(message.getSeqNumber());
			this.scheduleUpdateNeighborParams(id, message.getCost(), 
					!message.getConsume(), message.getRestoreTime());
		} else {
			nm.setMetrics(message.getCpu(), message.getMem(), message.getBW(),
					message.getStorage(), message.getAppIDs(), message.getMinRatio());
			nm.setSeqNumber(message.getSeqNumber());
		}

		nodesReputation.get(id).updateAlternative(message.getAlt());
		gridP2P.routeMyMsg(id, new ACKMsg(node.getLocalNodeHandle()));		
	}

	public void messageReceived(Id id, ACKMsg message) {
		log.write(this, "ACK received (id: " + id +
				", message: " + message + ")", Log.FINER);
				
		Id source = message.getSource();
		if(!updateTime.containsKey(source)) return;
		long startTick = updateTime.get(source);
		long endTick = System.currentTimeMillis();
		long diff = endTick - startTick; 
		log.write(this, "diffTime: " + diff, Log.FINER);
		
		//System.out.println("time diff: " + diff);
		
	}
	
	public void messageReceived(Id id, AvailabilityMsg message) {
		Id source = message.getSource();
		
		switch(message.getStatus()) {
		case AvailabilityMsg.IS_AVAILABLE:
			if(gridP2P.controlInbound(message))
				message.setStatus(AvailabilityMsg.AVAILABLE);
			else
				message.setStatus(AvailabilityMsg.NOT_AVAILABLE);
			message.setSource(node.getLocalNodeHandle());
			gridP2P.routeMyMsg(source, message);
			break;
		case AvailabilityMsg.AVAILABLE:
			synchronized(availableLock) {
				nAvailableNodes++;
				if(--waitingNReplies <= 0) {
					synchronized(this) {
						this.notify();
					}
				}
			}
			if(neighborsRate.containsKey(source))
				neighborsRate.get(source).setAvailability(true);
			break;
		case AvailabilityMsg.NOT_AVAILABLE:
			synchronized(availableLock) {
				notAvailableNodes++;
				if(--waitingNReplies <= 0)
					synchronized(this) {
						this.notify();
					}
			}
			if(neighborsRate.containsKey(source))
				neighborsRate.get(source).setAvailability(false);
			break;
		}
	}
	
	/**
	 * Updates the list (neighborsRate/nodeReputation) of neighbors closest to the node
	 * 
	 * @return boolean - true in case there were changes in the network
	 */
	public void joinNode(NodeHandle handle) {
		log.write(this, "joinNode called (handle: " + handle + ")", 10);
		
		Id id = handle.getId();

		if(neighborsRate.containsKey(id))
			return;
		
		double prox = gridP2P.getEndpoint().proximity(handle);

		neighborsRate.put(id, new NodeMetrics(handle, prox));
		nodesReputation.put(id, new NodeReputation(id, NUM_RETRIES));
		sendUpdateMsg(handle.getId(), buildUpdateMsg());
	}

	public void leaveNode(NodeHandle handle) {
		log.write(this, "leaveNode called (handle: " + handle + ")", Log.FINER);
		Id id = handle.getId();

		this.neighborsRate.remove(id);
		this.nodesReputation.remove(id);
	}
	
	private void initializeNeighborsRecon() {
		log.write(this, "initializeNeighborsRecon called", Log.FINE);
		
		List<NodeHandle> lst = gridP2P.getEndpoint().networkNeighbors(NUM_NEIGHBORS);
		if(lst.isEmpty()) {
			log.write(this, "No neighbors found", Log.FINE);
			networkInitialized = true;
			return;
		}
		
		for(NodeHandle nh : lst) {
			Id id = nh.getId();

			int prox = gridP2P.getEndpoint().proximity(nh);
			this.neighborsRate.put(id, new NodeMetrics(nh, prox));
			this.nodesReputation.put(id, new NodeReputation(id, NUM_RETRIES));
		}

		networkInitialized = true;
		sendUpdateMessages();    
		synchronized (updatesBuffer) {
			for(Id id : updatesBuffer.keySet()) {
				List<UpdateMsg> buffer = updatesBuffer.get(id);
				while(!buffer.isEmpty())
					messageReceived(id, buffer.remove(0));
			}
			updatesBuffer.clear();
		}
	}
	
	public UpdateMsg buildUpdateMsg(){
		log.write(this, "buildUpdateMsg called", 10);
				
		Message message = new UpdateMsg(
				gridP2P.getEndpoint().getLocalNodeHandle(), // This node handle - source
				resourceManif.getCPU(), resourceManif.getMem(), resourceManif.getBW(),
				resourceManif.getStorage(),
				gridP2P.getAppIDs(),
				false, // Consume?
				areAnyReliableNodes(), // Alternative
				seqNumber); // Sequence Number
		gridP2P.controlOutbound(message);
		return (UpdateMsg)message;
	}

	/**
	 * 
	 * A message (task) was received. If processed, the availability
	 * (payCost()) will be updated.
	 * 
	 * @param id - sender's id
	 * @param message - task
	 */
	public void messageReceived(Id id, Task message, long currentTime) {
		log.write(this, "task received (id: " + id + ", message: " + 
				message + ", currentTime: " + currentTime + ")", Log.FINER);
		
		if(gridP2P.controlInbound(message))
			gridP2P.processTask(id, message, currentTime);
	}	
	
	public void messageReceived(Id id, TaskResult message, long currentTime) {
		log.write(this, "taskResult received (id: " + id + ", message: " + 
				message + ", currentTime: " + currentTime + ")", 10);
		gridP2P.gatherTaskResult(id, message, currentTime);		
	}
	
	/**
	 * Removes the restriction imposed onto this node
	 * @param source - node's id
	 */
	private void clearRestrictions(Id source) {
		log.write(this, "clearRestrictions called (IdSource: " + source + ")", 10);

		for(List<Id> lst : chosenAvailableNodes.values())
			lst.remove(source);
	}

	/**
	 * Forwards an Update Message to all neighbors
	 */
	private void sendUpdateMessages() {
		log.write(this,"sending update messages to all neighbors", Log.FINER);
		Message msg = buildUpdateMsg();
		for(Id id : neighborsRate.keySet())	
			sendUpdateMsg(id, msg);		
	}
	
	private void sendUpdateMsg(Id id, Message message) {
		updateTime.put(id, System.currentTimeMillis());				
		gridP2P.routeMyMsg(id, message);
	}
	

	// periodically send update messages
	public void run() {
		while(true) {
			try {
				Thread.sleep(updateInterval);
			} catch (InterruptedException e) {
			}
			sendUpdateMessages();
		}
	}	
	
	/***************************************/
	/*** NodesReputation related methods ***/
	/***************************************/
	private synchronized double areAnyReliableNodes() {
		log.write(this, "areAnyReliableNodes called", 10);
		double best = 100;

		synchronized(nodesReputation) {
			for(NodeReputation rep : nodesReputation.values()) {
				// verify if it was chosen in the past, at least once
				if (rep.getChoosen() > 0 && rep.getRate() < best) {
					best = rep.getRate();
				}
			}
		}
		return best;
	}
	
	// Updates the reputation fields of the node 'id'
	// metric -> 0=Fails, 1=Retries, 2=BackToOrigin, 3=Alternative, 4=Chosen
	public void updateReputation(Id id, int metric, int value) {
		log.write(this, "updateReputation called (id: " + id + ", metric: " + metric + 
				", value: " + value + ")", 10);
		synchronized(neighborsRate) {
			NodeReputation rep = null;
			try{
				rep = (NodeReputation) nodesReputation.get(id);
				if(rep == null)
					return;
			}catch(NullPointerException e){
				log.write(this, "NullPointerException occurred: " + 
						e.getMessage(), Integer.MAX_VALUE);
			}

			switch (metric) {
			case 0: {
				rep.updateFails(value);
				break;
			}
			case 1: {
				rep.updateRetries(value);
				break;
			}
			case 2: {
				rep.updateBackToOrigin(value);
				break;
			}
			case 3: {
				rep.updateAlternative(value);
				break;
			}
			case 4: {
				rep.updateChoosen(value);
				break;
			}
			}
		}
	}

	/**
	 * Inserts the result of the processed task into the overlay using the
	 * Past - a large-scale, peer-to-peer archival storage facility.
	 * 
	 * @param key - PastContent object Id
	 * @param result - data to store in the overlay
	 */
	public void cacheTaskResult(Id key, final byte[] result) {
		log.write(this, "cacheTaskResult called (key: " + key.toStringFull() + 
				", result: " + result + ")" , Log.FINER);	

		// build the past content		
		final PastContent dataResult = new ContentResult(key, result);

		// insert the data
		pastApp.insert(dataResult, new Continuation<Boolean[], Exception>() {
			public void receiveResult(Boolean[] result) {
				log.write("receivedResult called", Log.FINER);
			}

			public void receiveException(Exception result) {
				log.write("Error storing on Past " + result.getMessage(), 
						Log.SEVERE);
			}
		});
	}

	public byte[] lookupCache(final Id lookupKey) {
		log.write(this, "lookupCache called (lookupKey: " + lookupKey, Log.FINER);
		resultDone = false;		
		pastApp.lookup(lookupKey, new Continuation<PastContent, Exception>() {
			public void receiveResult(PastContent content) {
				log.write("Successfully looked up " + content + 
						" for key " + lookupKey, Log.FINE);
				result = ((ContentResult)content).getTaskResult();
				resultDone = true;
			}

			public void receiveException(Exception result) {
				log.write("Error looking up " + lookupKey, Log.FINER);
				result = null;
				resultDone = true;				
			}
		});
		while(!resultDone) {
			try {
				Thread.sleep(1);
			} catch (InterruptedException e) {
			}
		}
		return result;
	}
	
	public long getMaxStorage() {
		return maxStorage;
	}
	
	public void setMaxStorage(long size) {
		maxStorage = size;
	}
	
	// getter for the freePastry environment
	public Environment getEnvironment() {
		return env;
	}
	
	// getter for the node
	public PastryNode getNode() {
		return node;
	}
	
	// Testing purposes --->
	public String getNeighborsDescription() {
		String str = "";
		for(NodeMetrics nm : neighborsRate.values())
			str = str + nm.toString() + '\n';
		return str;
	}
	
	public String getResourceManifest() {
		return resourceManif.toString();
	}
	// <--- end of testing methods
}
